package org.chen.ctrip.rmq.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.chen.ctrip.rmq.listener.AiProcessingMessageListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class RocketMQConfig {
    
    // 主题和消费者组常量
    public static final String AI_PROCESSING_TOPIC = "ai-processing-topic";
    public static final String AI_PROCESSING_CONSUMER_GROUP = "ai-processing-consumer-group";
    public static final String AI_RETRY_TOPIC = "ai-processing-retry-topic";
    public static final String AI_DLQ_TOPIC = "ai-processing-dlq-topic";
    
    /**
     * 生产者配置
     */
    @Bean
    public DefaultMQProducer aiProcessingProducer() {
        DefaultMQProducer producer = new DefaultMQProducer("ai-processing-producer-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.setRetryTimesWhenSendFailed(3);
        producer.setRetryTimesWhenSendAsyncFailed(3);
        producer.setSendMsgTimeout(10000);
        producer.setMaxMessageSize(4 * 1024 * 1024); // 4MB
        
        try {
            producer.start();
            log.info("RocketMQ AI处理生产者启动成功");
        } catch (MQClientException e) {
            log.error("RocketMQ AI处理生产者启动失败", e);
            throw new RuntimeException("RocketMQ生产者启动失败", e);
        }
        
        return producer;
    }
    
    /**
     * 消费者配置
     */
    @Bean
    public DefaultMQPushConsumer aiProcessingConsumer(AiProcessingMessageListener messageListener) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(AI_PROCESSING_CONSUMER_GROUP);
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(1); // 逐个处理，控制AI并发
        consumer.setConsumeThreadMin(4); // 最小消费线程数
        consumer.setConsumeThreadMax(8); // 最大消费线程数，控制AI并发
        consumer.setPullBatchSize(1); // 每次拉取1条消息
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式
        
        try {
            consumer.subscribe(AI_PROCESSING_TOPIC, "*");
            consumer.setMessageListener(messageListener);
            consumer.start();
            log.info("RocketMQ AI处理消费者启动成功");
        } catch (MQClientException e) {
            log.error("RocketMQ AI处理消费者启动失败", e);
            throw new RuntimeException("RocketMQ消费者启动失败", e);
        }
        
        return consumer;
    }
}